package com.rrd.dw.mr.gzip.kafka;

import com.alibaba.fastjson.JSONObject;
import com.euler.commons.web.JsonMapperBuilder;
import com.rrd.dw.utils.DateUtil;
import com.rrd.dw.utils.HadoopUtils;
import com.ucredit.babel.common.kafka.PostRecord;
import com.ucredit.babel.common.model.BaseModel;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.regex.Pattern;

/**
 * @author xj
 * @version 回溯三方数据，修复遗漏数据
 */
public class TongdunRecallMR extends Configured implements Tool {
    public static final String CRT_FLAG = "\001";
    public static final Pattern CRT_PATTERN = Pattern.compile(CRT_FLAG);
    private static final List<ParseBase> ParserList = getParseList();

    private static List<ParseBase> getParseList() {
        ArrayList<ParseBase> parsers = new ArrayList<>();
        parsers.add(new TongDunRiskHitRulesV2Parse());
        parsers.add(new TongDunRiskHitDimV2Parse());
        parsers.add(new TongDunRiskV2Parse());
        parsers.add(new TongDunRiskConditionsV2Parse());
        return parsers;
    }


    private static Map<String, String> getOutPath() {
        Map<String, String> map = new HashMap<String, String>();
        for (ParseBase parser : ParserList) {
            map.put(parser.getType() + parser.getSubType(), "/data/fdm/" + parser.getTableName() + "/dt=");
        }
        return map;
    }

    public static class CallMap extends Mapper<LongWritable, Text, Text, Text> {
        private Map<String, List<ParseBase>> parserListMap = new HashMap<>();

        @Override
        protected void setup(Mapper<LongWritable, Text, Text, Text>.Context context)
                throws IOException, InterruptedException {
            setParserMap();
        }

        private void setParserMap() {
            for (ParseBase parser : getParseList()) {
                List<ParseBase> parsers;
                if (parserListMap.containsKey(parser.getType())) {
                    parsers = parserListMap.get(parser.getType());
                } else {
                    parsers = new ArrayList<>();
                }
                parsers.add(parser);
                parserListMap.put(parser.getType(), parsers);
            }
        }

        public static String ToUnderline(String name) {
            StringBuilder result = new StringBuilder();
            if (name != null && name.length() > 0) {
                for (int i = 0; i < name.length(); i++) {
                    String s = name.substring(i, i + 1);
                    if (s.equals(s.toUpperCase()) && !Character.isDigit(s.charAt(0))) {
                        result.append("_");
                    }
                    result.append(s.toLowerCase());
                }
            }
            return result.toString();
        }

        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context)
                throws IOException, InterruptedException {
            if (value == null || value.toString().equals("")) {
                return;
            }

            String[] valList = value.toString().split(CRT_FLAG);
            JSONObject obj = JSONObject.parseObject(valList[2]);
            obj.put("type", "TONGDUN_RISK_RECORD");
            Long time = obj.getLong("timestamp");
            if(time == null){
                String queryTime = obj.getString("queryTime");
                /*hh_20210821161646_183375391549148624*/
                if(queryTime == null){
                    return;
                }
                time = DateUtil.formateToTimestamp(queryTime);
                obj.put("timestamp", time);
            }
            /*回溯2021-07-31之前的*/
            Long start = new Long("1627660800000");
            if(obj.getLong("timestamp") >= start){
                return;
            }

            String loanKey = obj.getString("loanKey");
            String userKey = obj.getString("userKey");
            if(userKey == null){
                if(valList[0] == null || valList[0].equals("")){
                    System.out.println(value.toString());
                    return;
                }
                obj.put("userKey", valList[0]);
            }
            if(loanKey == null){
                if(valList[1] == null || valList[1].equals("")){
                    System.out.println(value.toString());
                    return;
                }
                obj.put("loanKey", valList[1]);
            }
            JSONObject job1 = obj.getJSONObject("job");
            if(job1 == null){
                JSONObject job = new JSONObject();
                job.put("finished", true);
                job.put("requested", true);
                job.put("successful", true);
                job.put("startTime", time);
                job.put("endTime", time);
                job.put("systemID", "HAO_HUAN");
                obj.put("job", job);
            }

            /*t驼峰转下划线*/
            ArrayList<String> Names = new ArrayList<String>();
            Names.add("hitRuleResult");
            Names.add("reasonCode" );
            Names.add("reasonDesc");
            Names.add("ruleId");
            Names.add("resultsForDim");
            Names.add("eventType");
            Names.add("hitsForDim");
            Names.add("industryDisplayName");
            Names.add("originalDimType");
            Names.add("matchDimType");
            Names.add("fraudResult");
            Names.add("finalScore");
            Names.add("policySetName");
            Names.add("riskType");
            Names.add("policyName");
            Names.add("hitRulesConvert");
            Names.add("finalDecision");
            Names.add("policySet");
            Names.add("seqId");
            Names.add("spendTime");
            Names.add("backId");
            Names.add("policyUuid");
            Names.add("policyScore");
            Names.add("policyDecision");
            Names.add("policyMode");
            Names.add("dimValue");
            Names.add("dimType");
            Names.add("addressBValue");
            Names.add("addressAValue");
            Names.add("addressB");
            Names.add("addressA");
            Names.add("hitRules");
            Names.add("deviceInfo");
            Names.add("geoIpInfo");
            Names.add("codeDisplayName");
            Names.add("proxyIpType");
            Names.add("gpsA");
            Names.add("gpsB");
            Names.add("diffDisplay");
            Names.add("matchSubDimType");
            Names.add("subDimType");
            Names.add("calcType");
            Names.add("calcDimTypeValue");
            Names.add("calcDimType");
            Names.add("unitDisplayName");
            Names.add("parentUuid");
            Names.add("abnormalTags");
            Names.add("blackCntOne");
            Names.add("blackCntTwo");
            Names.add("blackCnt");
            Names.add("blackDst");
            Names.add("blackRat");
            Names.add("codeDisplayName");
            Names.add("coreDst");
            Names.add("coreNode");
            Names.add("emulatorType");
            Names.add("fraudDistOne");
            Names.add("fraudDistTwo");
            Names.add("fraudDist");
            Names.add("greyCnt");
            Names.add("greyRat");
            Names.add("groupId");
            Names.add("hookAddress");
            Names.add("hookIMP");
            Names.add("hookInline");
            Names.add("hookMethod");
            Names.add("nodeDist");
            Names.add("nodeScore");
            Names.add("replaceText");
            Names.add("totalCntTwo");
            Names.add("totalCnt");
            Names.add("evidenceTime");
            Names.add("overdueAmount");
            Names.add("overdueCounts");
            Names.add("overdueCount");
            Names.add("overdueDays");
            Names.add("overdueDay");
            Names.add("overdueTime");
            Names.add("repayStatus");
            Names.add("subjectType");
            Names.add("listStr");

            String val = obj.toJSONString();
            for(String k: Names){
                val = val.replace(k, ToUnderline(k));
            }
            String type = obj.getString("type");
            JSONObject input = new JSONObject();
            input.put("type", type);
            input.put("dataJson", val);
            input.put("action", "PUT");
            input.put("resendTimes", 0);
            input.put("md5", "");

            List<ParseBase> parsers = parserListMap.get(type);
            if (parsers != null && parsers.size() > 0) {
                String i = input.toJSONString();
                PostRecord r = JsonMapperBuilder.build().readValue(i, PostRecord.class);
                BaseModel<?> model = (BaseModel<?>) r.getRecord();
                if (model == null) {
                    System.out.println(i);
                }
                if (r.getAction().equals("PUT")) {
                    for (ParseBase parser : parsers) {
                        parser.setPartition("," + DateUtil.toDateStr(new Date(time)));
                        parseValue(model, parser, context);
                    }
                }
            }
        }

        protected void parseValue(BaseModel<?> model, ParseBase parser, Context context) throws IOException, InterruptedException {
            String successful = "1";
            if (model != null) {
                if (parser.parseJob(context, model, successful)) {
                    context.getCounter("counter", "user." + parser.getType() + parser.getSubType()).increment(1);
                } else {
                    context.getCounter("counter", "user." + parser.getType() + parser.getSubType() + ".false").increment(1);
                }
            } else {
                context.getCounter("error", "user." + parser.getType() + parser.getSubType() + ".null").increment(1);
            }
        }
    }

    public static class RecallReduce extends Reducer<Text, Text, Text, Text> {
        private MultipleOutputs<Text, Text> mos;
        private NullWritable key2 = NullWritable.get();
        private String dt = null;
        private Map<String, String> map = null;

        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            this.mos = new MultipleOutputs<Text, Text>(context);
            Configuration conf = context.getConfiguration();
            this.map = getOutPath();
        }

        @Override
        protected void cleanup(Context context) throws IOException, InterruptedException {
            this.mos.close();
        }

        private String keystr = null;

        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context)
                throws IOException, InterruptedException {
            keystr = key.toString();
            String[] keyArr = keystr.split(",");
            System.out.println("REDUCE KEY:" + keystr + "|" + this.map.get(keyArr[0]) + keyArr[1]);
            for (Text val : values) {
                context.getCounter("reduce", "reduce.success").increment(1);
                this.mos.write("textinfo", key2, val, this.map.get(keyArr[0]) + keyArr[1] + "/");
            }

        }
    }

    @Override
    public int run(String[] args) throws Exception {
        Configuration conf = getConf();
        conf.set("mapreduce.output.fileoutputformat.compress", "true");
        conf.set("mapreduce.output.fileoutputformat.compress.type", "BLOCK");
        conf.set("mapreduce.output.fileoutputformat.compress.codec", "org.apache.hadoop.io.compress.GzipCodec");
        conf.set("io.compression.codecs", "org.apache.hadoop.io.compress.GzipCodec");
        Job job = Job.getInstance(conf);
        String dt = conf.get("dt");
        String i = conf.get("i", "0");
        String output = "/tmp/bdm/tongdun_recall" + i;
        String inputDir = "/user/hive/warehouse/dwtmp.db/tongdun_distinct";
        System.out.println("param.input.path=" + inputDir);
        System.out.println("param.out.path=" + output);
//        Map<String, String> outputMap = getOutPath(dt);
//        for (String outputPath: outputMap.values()){
//            HadoopUtils.delete(conf, new Path(outputPath));
//        }

        Path outpath = new Path(output);
        HadoopUtils.delete(conf, outpath);
        Path[] inputPaths = HadoopUtils.getDirPaths(conf, new Path(inputDir)).toArray(new Path[0]);
        for (Path p: inputPaths) {
            System.out.println("input path:" + p);
        }
        TextInputFormat.setInputPaths(job, inputPaths);
        TextOutputFormat.setOutputPath(job, outpath);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);
        job.setOutputKeyClass(NullWritable.class);
        job.setOutputValueClass(Text.class);
        job.setJarByClass(TongdunRecallMR.class);
        job.setMapperClass(TongdunRecallMR.CallMap.class);
        job.setReducerClass(TongdunRecallMR.RecallReduce.class);
        MultipleOutputs.addNamedOutput(job, "textinfo", TextOutputFormat.class, Text.class, Text.class);
        int stat = job.waitForCompletion(true) ? 0 : 1;
        for (ParseBase parser : ParserList) {
            System.out.println(parser.getComment() + parser.getSubType() + ":" + job.getCounters().getGroup("counter").findCounter("user." + parser.getType() + parser.getSubType()).getValue());
            System.out.println(parser.getComment() + parser.getSubType() + "查询is Null次数：" + job.getCounters().getGroup("error").findCounter("user." + parser.getType() + parser.getSubType() + ".null").getValue());
        }
        System.out.println("異常次數：" + job.getCounters().getGroup("error").findCounter("ClassCastException").getValue());
        System.out.println("reduce.null次数：" + job.getCounters().getGroup("reduce").findCounter("reduce.null").getValue());
        System.out.println("reduce.success次数：" + job.getCounters().getGroup("reduce").findCounter("reduce.success").getValue());
        return stat;
    }

    public static void main(String[] args) throws Exception {
        int res = ToolRunner.run(new TongdunRecallMR(), args);
        System.exit(res);
    }
}
